package com.testmq;
 
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @ClassName TransactionListenerImpl
 * @Description TODO 事务消息监听器
 */
public class TransactionListenerImpl implements TransactionListener {
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object o) {
        // 执行订单本地事务
        // 接着执行本地一连串事务执行结果，去选择执行commit or rollback
        try {
            //doLocalSomeService()...
            // 如果本地事务都执行成功了，返回commit
            localTrans.put(msg.getTransactionId(), 1);
            return LocalTransactionState.COMMIT_MESSAGE;
        }catch (Exception e){
            // 本地事务执行失败，回滚所有一切执行过的操作
            // 如果本地事务执行失败了，返回rollback，标记half消息无效
            localTrans.put(msg.getTransactionId(), 2);
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
 
    // 如果因为各种原因，没有返回 commit 或者rollback
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 查询本地事务，是否执行成功了
        Integer status = localTrans.get(msg.getTransactionId());
        // 根据本地事务的情况去选择执行 commit or rollback
        if(null != status){
            switch (status){
                case 0: return LocalTransactionState.UNKNOW;
                case 1: return LocalTransactionState.COMMIT_MESSAGE;
                case 2: return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}